package com.shuhe.domain.mxgc;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONArray;
import com.alibaba.fastjson.JSONObject;
import com.shuhe.metastore.DBSql;
import com.shuhe.metastore.MetaMysql;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.sql.Connection;
import java.sql.ResultSet;
import java.util.*;

import static com.shuhe.common.CommonFunction.closeDb;


public class TaskExecute implements Runnable {

    JSONObject jsonObject;
    private static final Logger LOG = LoggerFactory.getLogger(TaskExecute.class);

    public TaskExecute(JSONObject jsonObject) {
        this.jsonObject = jsonObject;
    }


    public void run() {
        Connection metaConn = null;
        Connection dbConn = null;
        try {
            dbConn = DBSql.connect();
            metaConn = MetaMysql.connect();

            String task_pc = UUID.randomUUID().toString().replace("-", "");

            // 10.21
            String dbName = "";
            String tableName = "";
            String sqlContent = "";
            String sql = "SELECT DISTINCT a.task_id,b.id,b.user_name,b.db_name,b.table_name,b.table_comment,b.createtime," +
                    "b.sql_content,b.mxid,a.dest_table,b.input_tables,a.task_id FROM v_mxgc_user_task_tables a,mxgc_user_table b " +
                    "WHERE a.mxid=b.mxid AND a.task_id='" + jsonObject.getString("task_id") + "'";
            ResultSet rs = MetaMysql.querySql(metaConn, sql);
            while (rs.next()) {
                dbName = rs.getString("db_name");
                tableName = rs.getString("table_name");
                sqlContent = rs.getString("sql_content");
            }
            taskExceute(metaConn, dbConn, task_pc, jsonObject.getString("task_id"),
                    dbName, tableName, sqlContent, "runImmedia");

        } catch (Exception ex) {
            ex.printStackTrace();
        } finally {
            closeDb(dbConn);
            closeDb(metaConn);
        }
    }

    public static void taskExceute(Connection metaConn, Connection dbConn, String batchId, String taskId, String dbName,
                                   String table_name, String sqlContent, String task_rule) throws Exception {
        String sql = "";
        try {

            writeLogMx(metaConn, "", "", batchId, taskId, task_rule);
            //获取增量相关信息
            String table_update_type = ""; //是否增量，
            String table_update_name = ""; //合并字段
            String table_update_column = "";
            String table_update_relation = "";
            Map<String, String> mapTableUpdates = new HashMap<>();
            mapTableUpdates = getModelTableUpdate(metaConn, taskId);
            for (Map.Entry<String, String> entry : mapTableUpdates.entrySet()) {
                if (entry.getKey().equals("table_update_type")) {
                    if (entry.getValue() != null || !entry.getValue().equals("")) {
                        table_update_type = entry.getValue();
                    }
                }
                if (entry.getKey().equals("table_update_name")) {
                    if (entry.getValue() != null || !entry.getValue().equals("")) {
                        table_update_name = entry.getValue();
                    }
                }
                if (entry.getKey().equals("table_update_column")) {
                    if (entry.getValue() != null || !entry.getValue().equals("")) {
                        table_update_column = entry.getValue();
                    }
                }
                if (entry.getKey().equals("table_update_relation")) {
                    if (entry.getValue() != null || !entry.getValue().equals("")) {
                        table_update_relation = entry.getValue();
                    }
                }
            }
            // 列映射
            List<TableUpdateRelation> tableUpdateRelations = JSON.parseArray(table_update_relation,
                    TableUpdateRelation.class);
            String destTable = dbName + "." + table_name;

            //主方法，分两种情况

            //全量
            sql = " truncate table  " + destTable;
            DBSql.execSql(dbConn, sql);
            sql = "insert into  " + destTable + " " + sqlContent;
            DBSql.execSql(dbConn, sql);

            //增量
            if (table_update_type.equals("true")) {
                //增量
                //计算新数据产生的增量时间，并删除
                String time = "";
                String columnStr = "";
                String columnStrHb = "";
                ArrayList<String> columns = new ArrayList<>();
                //处理空的情况,如果为空，就不用执行了
                int cun = 0;
                sql = "select count(" + table_update_column + ") cun from (" + sqlContent + ") a";
                ResultSet rs = DBSql.querySql(dbConn, sql);
                while (rs.next()) {
                    cun = rs.getInt(1);
                }
                sql = "select " + table_update_column + " from (" + sqlContent + ") a";
                rs = DBSql.querySql(dbConn, sql);
                while (rs.next()) {
                    time = rs.getString(table_update_column);
                }
                String table_update_column_hb = "";
                for (TableUpdateRelation relation : tableUpdateRelations) {
                    if (relation.getColumnName().equals(table_update_column)) {
                        table_update_column_hb = relation.getColumnName1();
                    }
                }
                if (cun > 0) {
                    sql = "delete from " + table_update_name + " where " + table_update_column_hb + "='" + time + "'";
                    DBSql.execSql(dbConn, sql);
                }
//                sql = "select column_name from mxgc_user_column where concat(db_name,'.',table_name)='" +
//                table_update_name + "' and upper(column_name) not in('INNER_ID','INNER_TIME')";
                sql = "select a.column_name from standard_column a,standard_table b where a.table_id=b.id and concat" +
                        "(b.table_type,'.',b.table_name)='" + table_update_name + "' and upper(a.column_name) not in" +
                        "('INNER_ID','INNER_TIME')";
                rs = MetaMysql.querySql(metaConn, sql);
                while (rs.next()) {
                    String column_name = rs.getString("column_name");
                    columns.add(column_name);
                }
                int j = 0;
                for (TableUpdateRelation relation : tableUpdateRelations) {
                    String column_name = "";
                    String column_name1 = "";

                    for (int i = 0; i < columns.size(); i++) {
                        if (relation.getColumnName1().equals(columns.get(i))) {
                            column_name1 = relation.getColumnName1();
                            column_name = relation.getColumnName();
                            break;
                        }
                    }
                    if (j == 0) {
                        columnStrHb = column_name1 + ",";
                        columnStr = column_name + ",";
                        j++;
                    } else {
                        columnStrHb = columnStrHb + column_name1 + ",";
                        columnStr = columnStr + column_name + ",";
                        j++;
                    }
                }
                columnStrHb = columnStrHb.substring(0, columnStrHb.length() - 1); //本表
                columnStr = columnStr.substring(0, columnStr.length() - 1);    //合并数据集
                if (cun > 0) {
                    sql = "insert into " + table_update_name + "(" + columnStrHb + ") select " + columnStr + " from " +
                            "(" + sqlContent + ") a";
                    DBSql.execSql(dbConn, sql);
                }
            }
            //结束日志
            sql = "select count(*) cun from " + destTable;
            String jls = "";
            ResultSet rs = DBSql.querySql(dbConn, sql);
            if (rs.next()) {
                jls = rs.getString("cun");
            }
            writeLogMx(metaConn, "", jls, batchId, taskId, task_rule);
            Thread.sleep(1000);

        } catch (Exception ex) {
            ex.printStackTrace();
            writeLogMx(metaConn, sql + ex.getMessage(), "", batchId, taskId, task_rule);
            LOG.error("失败:" + sql, ex);
        }
    }

    public static HashMap<String, String> getModelTableUpdate(Connection metaConn, String taskId) throws Exception {
        HashMap<String, String> tableUpdates = new HashMap<>();
        String sql =
                "select b.mxnr from mxgc_user_task a,mxgc_user_mxlb b where a.mx_id=b.id and a.task_id='" + taskId +
                        "'";
        ResultSet rs = MetaMysql.querySql(metaConn, sql);
        while (rs.next()) {
            String mxnr = rs.getString("mxnr").replace("\\\"", "\"");
            JSONObject json = JSONObject.parseObject(mxnr);
            JSONArray cells = json.getJSONArray("cells");
            for (int i = 0; i < cells.size(); i++) {
                JSONObject cell = cells.getJSONObject(i);
                String shape = cell.getString("shape");
                if (shape.equals("edge")) {
                    continue;
                }
                JSONObject data = cell.getJSONObject("data");
                String table_update_relation = data.getString("table_update_relation");
                String table_update_name = data.getString("table_update_name");
                String table_update_column = data.getString("table_update_column");
                String table_update_type = data.getString("table_update_type");
                if (table_update_relation != null) {
                    tableUpdates.put("table_update_relation", table_update_relation);
                }
                if (table_update_name != null) {
                    tableUpdates.put("table_update_name", table_update_name);
                }
                if (table_update_column != null) {
                    tableUpdates.put("table_update_column", table_update_column);
                }
                if (table_update_type != null) {
                    tableUpdates.put("table_update_type", table_update_type);
                }
            }
        }
        return tableUpdates;
    }

    //模型记录日志
    public static void writeLogMx(Connection metaConn, String error_msg,
                                  String record_count, String batchId, String taskId, String taskRule) {
        error_msg = error_msg.replaceAll("'", "''");
        if (error_msg.length() > 4000) {
            error_msg = error_msg.substring(0, 4000);
        }
        String sql = "select task_pc, task_id from mxgc_user_task_log where task_pc = '" + batchId + "' " +
                "and" + "  " + "task_id = '" + taskId + "' and task_rule='" + taskRule + "'";
        try {
            ResultSet rs = MetaMysql.querySql(metaConn, sql);
            if (rs.next()) {
                String run_status = "";
                if (error_msg.length() > 0) {
                    run_status = "fail";
                } else {
                    run_status = "success";
                }
                if (run_status.equals("success")) {
                    sql = "update mxgc_user_task_log set  zzsj = CURRENT_TIMESTAMP, " +
                            "error_msg " + "= '" + error_msg + "', sjl = '" + record_count +
                            "',status='" + run_status + "' where task_pc =" +
                            " '" + batchId + "' and task_id = '" + taskId + "' and task_rule='" + taskRule + "'";
                } else {
                    sql = "update mxgc_user_task_log set  zzsj = '', " +
                            "error_msg " + "= '" + error_msg + "', sjl = '" + record_count +
                            "',status='" + run_status + "' where task_pc =" +
                            " '" + batchId + "' and task_id = '" + taskId + "' and task_rule='" + taskRule + "'";
                }
            } else {
                sql = "insert into mxgc_user_task_log(task_pc,task_id,task_name,task_rule,qssj,zzsj)\n" +
                        "select '" + batchId + "','" + taskId + "',(select task_name from mxgc_user_task a where "
                        + "a.task_id='" + taskId + "'),'" + taskRule + "'," +
                        "CURRENT_TIMESTAMP,''";
            }
            MetaMysql.execSql(metaConn, sql);
        } catch (Exception ex) {
            ex.printStackTrace();
            LOG.error("写入日志失败", ex);
        }
    }
}
